package com.g2room.queue.kafka;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;


public class KafkaConsumer extends Thread {

//	private String topic = "dota2_match_history";
	private String topic = "timetaskSeqDetail";
//	private String topic = KafkaProducer.topic;//KafkaProducer.topic;

	public KafkaConsumer() {
		super();
	}

	@Override
	public void run() {
		ConsumerConnector consumer = createConsumer();
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, 1); // 一次从主题中获取一个数据
		Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
		KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据
		ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
		int i = 0;
		Long start = System.currentTimeMillis();
//		int count = 1000;
		long count = 0;
		while (iterator.hasNext()) {
			String message = new String(iterator.next().message());
			System.out.println(count + "--" + (message.length()>21?message.substring(0, 22):message));
			count++;
//			i++;
//			if(i % count == 0) {
//				Long end = System.currentTimeMillis();
//				System.out.println("当前数据量=" + i + "，耗时：" + (end -start) + "，每秒数据：" + (count * 1000 / Math.max((end - start), 1)));
//				start = end;
//			}
		}
		
	}

	private ConsumerConnector createConsumer() {
		return Consumer.createJavaConsumerConnector(new ConsumerConfig(getProperties()));
	}
	
	
	public static Properties getProperties() {
		Properties properties = new Properties();
		properties.put("zookeeper.connect", "10.3.246.63:2180,10.3.246.60:2180");// zk
		/**
		 * kfk.topic=timetaskSeqDetail
kfk.zk.connect=10.3.246.60:2180
kfk.session.timeout.ms=4000
kfk.sync.time.ms=200
kfk.metadata.broker.list=10.3.246.63:9092,10.3.246.60:9092
kfk.request.required.acks=1
		 */
		// 必须要使用别的组名称，
		// 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据
		properties.put("group.id", "consumer-01");
		 //zk连接超时
		properties.put("zookeeper.session.timeout.ms", "4000");
		properties.put("zookeeper.sync.time.ms", "200");
		properties.put("auto.commit.interval.ms", "1000");
		properties.put("auto.offset.reset", "smallest");
        //序列化类
		properties.put("serializer.class", "kafka.serializer.StringEncoder");
		return properties;
	}

	public static void main(String[] args) {
		new KafkaConsumer().start();
	}

}